package cn.lhz.indiaStation.etl

import cn.lhz.util.spark.SparkUtil
import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.functions.col
import org.apache.spark.sql.types.{DoubleType, IntegerType}

/**
 * 对 global.csv 进行数据清洗，过滤掉包含空值的行
 *
 * @author 年阔正
 * @version 1.0.0
 */
object IndiaStationEtl {
  def main(args: Array[String]): Unit = {
    // 获取 SparkSession 对象
    val sparkSession = SparkUtil()

    // 定义源数据在HDFS文件系统的路径
    val originPath: String = "/air-data/ods/IndiaStationDay.csv"

    // 读取 CSV 文件，获取 DataFrame
    val df = sparkSession.read.option("header", "true").csv(originPath)

    // 字段集合
    val cols = Array(
      "stationId",       // 监测站点ID
      "date",            // 日期
      "pm2_5",           // PM2.5浓度
      "pm10",            // PM10浓度
      "no",              // 一氧化氮（NO）浓度
      "no2",             // 二氧化氮（NO2）浓度
      "nox",             // 氮氧化物（NOx）浓度
      "nh3",             // 氨气（NH3）浓度
      "co",              // 一氧化碳（CO）浓度
      "so2",             // 二氧化硫（SO2）浓度
      "o3",              // 臭氧（O3）浓度
      "benzene",         // 苯（Benzene）浓度
      "toluene",         // 甲苯（Toluene）浓度
      "xylene",          // 二甲苯（Xylene）浓度
      "aqi",             // 空气质量指数（AQI）
      "aqi_bucket"       // AQI等级区间
    )


    // 过滤掉有空值的数据行
    val cleanedDf = df.na.drop(cols)

    // 为字段指定数据类型
    val table = cleanedDf.select(
      col("stationId"),
      col("date"),
      col("pm2_5") cast (DoubleType),        // PM2.5浓度
      col("pm10") cast (DoubleType),         // PM10浓度
      col("no") cast (DoubleType),           // 一氧化氮（NO）浓度
      col("no2") cast (DoubleType),          // 二氧化氮（NO2）浓度
      col("nox") cast (DoubleType),          // 氮氧化物（NOx）浓度
      col("nh3") cast (DoubleType),          // 氨气（NH3）浓度
      col("co") cast (DoubleType),           // 一氧化碳（CO）浓度
      col("so2") cast (DoubleType),          // 二氧化硫（SO2）浓度
      col("o3") cast (DoubleType),           // 臭氧（O3）浓度
      col("benzene") cast (DoubleType),      // 苯（Benzene）浓度
      col("toluene") cast (DoubleType),      // 甲苯（Toluene）浓度
      col("xylene") cast (DoubleType),       // 二甲苯（Xylene）浓度
      col("aqi") cast (IntegerType),         // 空气质量指数（AQI）
      col("aqi_bucket")
    )

    // 保存数据到 Hive 表
    table.write.mode(SaveMode.Overwrite).saveAsTable("air.dwd_india_station")

    // 释放资源
    sparkSession.stop()
  }
}